/*tsdb related redis command implements */
#include "server.h"
#ifdef USE_TSDB

#ifndef _MAX_
#define _MAX_(a,b) ((a) > (b) ? a : b)
#endif //!_MAX_
#ifndef _MIN_
#define _MIN_(a,b) ((a) < (b) ? a : b)
#endif //!_MIN_

#define KMinTimestampDelta (1)
/* Convert hexstring into double value */
#define DOUBLE_SIZE 8

static int sdsHexStrToDouble(sds s, double *target) {
  size_t len = sdslen(s);
  if (len != DOUBLE_SIZE*2) {
    return C_ERR;
  }
  unsigned char dchar[DOUBLE_SIZE] = { 0 };
  char* pos = s;
  size_t count = 0;
  for (count = 0; count < sizeof(dchar) / sizeof(dchar[0]); count++) {
    sscanf(pos, "%2hhx", &dchar[count]);
		pos += 2 * sizeof(char);
  }
  void* t = (void*)dchar;
  *target = (*(double *)t);
  return C_OK;
}
/*
static int sdsHexStrToDoubleOrReply(client *c, sds s, double *target, const char *msg) {
  size_t len = sdslen(s);
  if (len != DOUBLE_SIZE*2) {
    // error format
    if (msg != NULL) {
        addReplyError(c,(char*)msg);
    } else {
        char errMsg[255] = {0};
        sprintf(errMsg, "invalid double len for hexstring %ld", len);
        addReplyError(c, errMsg);
    }

    return C_ERR;
  }

  unsigned char dchar[DOUBLE_SIZE] = { 0 };
  char* pos = s;
  size_t count = 0;
  for (count = 0; count < sizeof(dchar) / sizeof(dchar[0]); count++) {
    sscanf(pos, "%2hhx", &dchar[count]);
		pos += 2 * sizeof(char);
  }
  void* t = (void*)dchar;
  *target = (*(double *)t);
  return C_OK;
}

*/

static sds genTSInfoString(tstream *t) {
  sds info = sdsempty();
  double* p = (double*)&t->prev_value;
  info = sdscatprintf(info,
      "# Time series data\r\n"
      "first_timestamp:%ld\r\n"
      "prev_timestamp:%ld\r\n"
      "prev_timestamp_delta:%ld\r\n"
      "prev_value:%f\r\n"
      "prev_value_leading_zeros:%d\r\n"
      "prev_value_trailing_zeros:%d\r\n"
      "num_values:%ld\r\n"
      "num_bits:%d\r\n"
      "size:%ld\r\n",
      t->first_timestamp,
      t->prev_timestamp,
      t->prev_timestamp_delta,
      *p,
      t->prev_value_leading_zeros,
      t->prev_value_trailing_zeros,
      tstreamNumValues(t),
      t->num_bits,
      tstreamSize(t)
  );

  return info;
}

/* ------------------------------------------ command implements ---------------------- */
void taddCommand(client *c) {
  int i;
  int added = 0;
  robj *ts;

  if ((c->argc % 2) == 1) {
      addReplyError(c,"wrong number of arguments for TADD");
      return;
  }

  ts = lookupKeyWrite(c->db,c->argv[1]);
  if (ts == NULL) {
      ts = createTimeSeriesObject();
      dbAdd(c->db,c->argv[1],ts);
  } else {
      if (ts->type != OBJ_TSDB) {
          addReply(c,shared.wrongtypeerr);
          return;
      }
  }

  for (i = 2; i < c->argc; i += 2) {
    long long timestamp;
    double value;
    if (getLongLongFromObjectOrReply(c, c->argv[i], &timestamp, "invalid parameters") != C_OK) return;

    if (sdsHexStrToDouble(c->argv[i+1]->ptr, &value) != C_OK) {
      if (getDoubleFromObjectOrReply(c, c->argv[i+1], &value, NULL) != C_OK) {
        return;
      }
    }


    if (tstreamAppend(ts->ptr, timestamp, value, KMinTimestampDelta) == C_OK) {
      added++;
    }
  }

  server.dirty += added;
  addReplyLongLong(c,added);
}

void tgetCommand(client *c) {
  robj *ts;
  long long start, end, n;
  if ((ts=lookupKeyReadOrReply(c, c->argv[1], shared.nullmultibulk)) == NULL) {
    return;
  }

  if (ts->type != OBJ_TSDB) {
    addReply(c,shared.wrongtypeerr);
    return;
  }

  if (c->argc > 2) {
    if (getLongLongFromObjectOrReply(c, c->argv[2], &start, "invalid start input") != C_OK) {
      return;
    }
  } else {
    start = 0;
  }

  if (c->argc > 3) {
    if (getLongLongFromObjectOrReply(c, c->argv[3], &end, NULL) != C_OK) {
      return;
    }
  } else {
    end = -1;
  }

  if (c->argc > 4) {
    if (getLongLongFromObjectOrReply(c, c->argv[4], &n, NULL) != C_OK) {
      return;
    }
  } else {
    n = -1;
  }

  tsiter iter;
  tstreamInitIterator(&iter);
  tvalue tvFirst;
  tvFirst.timestamp = 0;
  tvFirst.value = 0;
  if (tstreamHasNext(&iter, ts->ptr) != C_OK) {
    // no value in ts
    addReply(c, shared.nullmultibulk);
    return;
  }

  if (tstreamNextValue(&iter, ts->ptr, &tvFirst) != C_OK) {
    // read first value error
    addReplyError(c,"read first tvalue error");
    return;
  }

  if (end > 0 && tvFirst.timestamp > end) {
    addReply(c, shared.nullmultibulk);
    return;
  }

  void *replylen = NULL;
  replylen = addDeferredMultiBulkLength(c);
  int count = 0;
  if (tvFirst.timestamp >= start) {
    // add first tvalue to result
    addReplyDouble(c, tvFirst.timestamp);
    addReplyDouble(c, tvFirst.value);
    count++;
  }
  while(tstreamHasNext(&iter, ts->ptr) == C_OK) {
    if (n >= 0 && count >= n ) break;

    tvalue tv;
    if (tstreamNextValue(&iter, ts->ptr, &tv) != C_OK) {
      break;
    }

    if (end > 0 && tv.timestamp > end) {
      break;
    }

    if (tv.timestamp >= start) {
      addReplyDouble(c, tv.timestamp);
      addReplyDouble(c, tv.value);
      count++;
    }
  }

  setDeferredMultiBulkLength(c, replylen, count*2);
}

void tinfoCommand(client *c) {
  robj *ts;
  if ((ts=lookupKeyReadOrReply(c, c->argv[1], shared.nullmultibulk)) == NULL) {
    return;
  }

  addReplyBulkSds(c, genTSInfoString((tstream*)ts->ptr));
}
#endif //!USE_TSDB
